package com.scala;

import org.apache.avro.generic.GenericData;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.function.Consumer;

public class Java_Lesson2 {
    public static void main(String args[]){
        SparkConf conf=new SparkConf();
        conf.setMaster("local").setAppName("lesson2");
        JavaSparkContext sc=new JavaSparkContext(conf);

        List<String> datas = Arrays.asList("zhangsan", "emp","lisi", "worker","wangwu", "teacher","zhaoliu", "friend");
        JavaRDD<String> rdd = sc.parallelize(datas,3);
        rdd.mapPartitionsWithIndex(new Function2<Integer, Iterator<String>, Iterator<String>>() {
            List<String> values=new ArrayList<>();
            @Override
            public Iterator<String> call(Integer index, Iterator<String> value) throws Exception {
                value.forEachRemaining(new Consumer<String>() {
                    @Override
                    public void accept(String s) {
                        System.out.println("分区索引index="+index+" 元素值为"+s);
                        values.add(s);
                    }
                });
                return values.iterator();
            }
        }, true)
                .repartition(4)
                .mapPartitionsWithIndex(new Function2<Integer, Iterator<String>, Iterator<String>>() {
            List<String> values=new ArrayList<>();
            @Override
            public Iterator<String> call(Integer index, Iterator<String> value) throws Exception {
                value.forEachRemaining(new Consumer<String>() {
                    @Override
                    public void accept(String s) {
                        System.out.println("重新分区之后的分区索引index="+index+" 元素值为"+s);
                        values.add(s);
                    }
                });
                return values.iterator();
            }
        }, true).collect();

        sc.close();
    }
}
